package com.yh.csx.event.store.gateway.event;

import com.google.common.eventbus.Subscribe;
import com.yh.csx.bsf.cat.method.CatMethod;
import com.yh.csx.bsf.core.util.JsonUtils;
import com.yh.csx.bsf.core.util.StringUtils;
import com.yh.csx.business.api.entity.Page;
import com.yh.csx.event.client.event.*;
import com.yh.csx.event.client.exception.EventErrorCode;
import com.yh.csx.event.client.exception.EventException;
import com.yh.csx.event.common.id.IdGenerator;
import com.yh.csx.event.common.utils.BeanUtils;
import com.yh.csx.event.common.utils.DateHelper;
import com.yh.csx.event.domain.event.EventStatusEnum;
import com.yh.csx.event.domain.gateway.IEventGateway;
import com.yh.csx.event.store.config.LogMsg;
import com.yh.csx.event.store.dynamic.DynamicQuery;
import com.yh.csx.event.store.event.dto.req.EventReq;
import com.yh.csx.event.store.event.dto.resp.AckEventResp;
import com.yh.csx.event.store.event.dto.resp.EventResp;
import com.yh.csx.event.store.gateway.event.dataobject.BizEventDo;
import com.yh.csx.event.store.gateway.event.dataobject.BizEventResultDo;
import com.yh.csx.event.store.gateway.project.ProductRepository;
import com.yh.csx.event.store.gateway.project.ProjectGateway;
import com.yh.csx.event.store.gateway.project.ProjectRepository;
import com.yh.csx.event.store.gateway.project.TeamRepository;
import com.yh.csx.event.store.gateway.subscribe.ProducerRepository;
import com.yh.csx.event.store.mapstruct.EventMapStruct;
import com.yh.csx.event.store.mapstruct.EventResultMapStruct;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.eventmesh.client.tcp.SimpleSubClient;
import org.apache.eventmesh.client.tcp.common.EventMeshCommon;
import org.apache.eventmesh.client.tcp.common.ReceiveMsgHook;
import org.apache.eventmesh.client.tcp.impl.LoadBalanceContext;
import org.apache.eventmesh.common.protocol.SubcriptionType;
import org.apache.eventmesh.common.protocol.SubscriptionMode;
import org.apache.eventmesh.common.protocol.tcp.Package;
import org.apache.eventmesh.common.protocol.tcp.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.domain.Example;
import org.springframework.data.domain.PageRequest;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.sql.Timestamp;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.yh.csx.event.client.Constants.*;
import static com.yh.csx.event.client.event.EventBusUtils.publish;
import static com.yh.csx.event.client.event.EventBusUtils.register;
import static com.yh.csx.event.client.event.EventMessageUtils.createPubEvent;

/**
 * @author zhangsheng
 * @date 2021/7/29 15:47
 */
@Slf4j
@Component
public class EventGateway implements IEventGateway, ReceiveMsgHook, CommandLineRunner {
    public static final String EVENT_CLIENT_ACK_TOPIC = "csx_b2b_event_client_ack_topic";
    public static final String EVENT_CONSUME_TOPIC = "event_sub_";
    @Autowired
    private EventRepository eventRepository;
    @Autowired
    private EventResultRepository eventResultRepository;
    @Autowired
    private LoadBalanceContext loadBalanceContext;
    @Autowired
    private EventResultMapStruct eventResultMapStruct;
    @Autowired
    private IdGenerator<String> idGenerator;
    @Autowired
    private ProductRepository productRepository;
    @Autowired
    private ProjectRepository projectRepository;
    @Autowired
    private TeamRepository teamRepository;
    @Autowired
    private ProducerRepository producerRepository;
    @Autowired
    @Qualifier("eventRetryTemplate")
    private RetryTemplate retryTemplate;
    @Autowired
    private EventMapStruct eventMapStruct;
    @Autowired
    private DynamicQuery dynamicQuery;
    @Value("${bsf.event.backup.before.days:2}")
    private int beforeDays;
    @Autowired
    private ProjectGateway projectGateway;

    public void init() {
        register(this);
    }

    @Override
    public Page<EventResp> queryEvents(EventReq req) {
        try {
            Stream<EventResp> eventRespStream = dynamicQuery.<Map>nativeQueryListMap(sql(req))
                    .stream()
                    .map(this::toEventResp);
            List<EventResp> list = eventRespStream
                    .collect(Collectors.<EventResp>toList());
            list.forEach(resp -> projectGateway.configName(resp));
            return new Page<EventResp>(req.getPage(), req.getSize(), count(req), list);
        } catch (ParseException e) {
            log.error("=queryEvents=>parseData : {} error .", req.getStartTime());
            throw new EventException(EventErrorCode.PARSE_DATA_FAIL, req.getStartTime());
        }
    }

    private int count(EventReq req) throws ParseException {
        List<Map> list = dynamicQuery.<Map>nativeQueryListMap(countSql(req));
        if (CollectionUtils.isEmpty(list)) {
            return 0;
        }
        return Integer.valueOf(list.get(0).get("count").toString());
    }

    private EventResp toEventResp(Map<String, Object> map) {
        return new EventResp()
                .setProjectId(Long.valueOf(null == map.get("projectId") ? "0" : map.get("projectId").toString()))
                .setEventId(null == map.get("eventId") ? "0" : map.get("eventId").toString())
                .setBizId(null == map.get("bizId") ? "0" : map.get("bizId").toString())
                .setBizType(null == map.get("bizType") ? "0" : map.get("bizType").toString())
                .setEventType(null == map.get("eventType") ? "0" : map.get("eventType").toString())
                .setCreator(null == map.get("creator") ? "0" : map.get("creator").toString())
                .setProjectName(null == map.get("projectName") ? "0" : map.get("projectName").toString())
                .setLocationCode(null == map.get("locationCode") ? "0" : map.get("locationCode").toString())
                .setRemark(null == map.get("remark") ? "0" : map.get("remark").toString())
                .setStackContext(null == map.get("stackContext") ? "0" : map.get("stackContext").toString())
                .setUpdateTime(null == map.get("updateTime") ? null : DateFormatUtils.format((Timestamp) map.get("updateTime"), YYYY_MM_DD_HH_MM_SS))
                .setStatus(EventStatusEnum.getEventStatusEnum(Integer.valueOf(null == map.get("status") ? "0" : map.get("status").toString())).getName())
                ;
    }

    @LogMsg
    @Override
    @CatMethod("event")
    public String rePush(String eventId) {
        BizEventDo bizEventDo = new BizEventDo();
        bizEventDo.setEventId(eventId);
        List<BizEventDo> list = eventRepository.findAll(Example.of(bizEventDo));
        if (CollectionUtils.isEmpty(list)) {
            log.error("=rePush=>Not fund msg by eventId:{}", eventId);
            throw new EventException(EventErrorCode.MAP_CONVERT_BEAN_FAIL, eventId);
        }
        BizEventDo eventDo = list.get(0);
        log.info("=rePush=>rePush topic:{} , msgId:{} , msg:{}", eventDo.getTopic(), eventId, eventDo);
        try {
            loadBalanceContext.getDefaultPub().publish(createPubEvent(eventMapStruct.convert2Message(eventDo)),
                    EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
        } catch (Exception e) {
            log.warn("=rePush=>rePush msgId:{}", eventId, e);
            throw new EventException(EventErrorCode.RE_PUSH_MSG_ERROR, eventId);
        }
        return eventDo.getEventBody();
    }


    @EventSubscribe(topic = EVENT_CLIENT_ACK_TOPIC, consumerGroup = "csx_b2b_event_sub_ack_event")
    public void mergeAckEvent(AckEventResp ackEvent) {
        BizEventResultDo convert = eventResultMapStruct.convert(ackEvent);
        convert.setId(idGenerator.nextId());
        if (StringUtils.isEmpty(convert.getTag())) {
            convert.setTag("no");
        }
        if (StringUtils.isEmpty(convert.getRemark())) {
            convert.setRemark("no");
        }
        if (StringUtils.isEmpty(convert.getCreator())) {
            convert.setCreator("sys");
        }
        if (EventStatusEnum.CONSUME_OK.getCode().equals(convert.getStatus())) {
            log.info("=saveAckEvent=>msg:{}", convert);
            eventResultRepository.mergeAckEvent(convert);
        } else {
            log.info("=saveUnAckEvent=>msg:{}", convert);
            mergeErrorEvent(convert);
        }
    }

    public void mergeErrorEvent(BizEventResultDo convert) {
        log.info("=saveErrorEvent=>msg:{}", convert);
        eventResultRepository.mergeErrorEvent(convert);
    }

    @Subscribe
    public void subscribe(ReloadEvent reloadEvent) {
        log.info("=subscribe=>reloadEvent:{}", reloadEvent);
        producerRepository.findAll().stream()
                .filter(config -> !config.getTopic().equals(EVENT_CLIENT_ACK_TOPIC))
                .distinct().forEach(config -> {
            SimpleSubClient client = loadBalanceContext.getSub(EVENT_CONSUME_TOPIC + config.getTopic());
            try {
                if (!client.isStartUp()) {
                    client.init();
                    client.heartbeat();
                }
                client.subscribe(config.getTopic(), SubscriptionMode.CLUSTERING, SubcriptionType.ASYNC);
                client.registerBusiHandler(this);
                client.listen();
                log.info("<=subscribe=>clientSubtopic:{} by consumeGroup:{} ,client:{}", config.getTopic(),
                        EVENT_CONSUME_TOPIC + config.getTopic(), client.toString());
            } catch (Exception e) {
                String warn = String.format("<=subscribe=>client sub topic:{} by consumeGroup:{} error.",
                        config.getTopic(), EVENT_CONSUME_TOPIC + config.getTopic());
                log.warn(warn, e);
                throw new EventException(warn);
            }
        });
    }


    @Override
    public void handle(Package pkg, ChannelHandlerContext ctx) {
        log.info("=handle=>pkg:{}", JsonUtils.serialize(pkg));
        EventMeshMessage msg = null;
        try {
            msg = (EventMeshMessage) pkg.getBody();
            BizEventDo event = eventMapStruct.convert(msg);
            if (StringUtils.isEmpty(event.getTag())) {
                event.setTag("no");
            }
            event.setProjectId(projectGateway.getCacheProjectByName(event.getAppName()).getId());
            eventRepository.mergeEvent(event);
            log.info("=handle=>topic:{} ,saveEvent:{}", event.getTopic(), event);
            publish(event);
        } catch (Exception e) {
            log.warn("=handle=>pkg:{}", JsonUtils.serialize(pkg), e);
            msg.getProperties().put(ACK_STATUS, String.valueOf(EventStatusEnum.CONSUME_ING.getCode()));
        }
    }

    @Override
    public void run(String... args) throws Exception {
        init();
    }

    public String sql(EventReq req) throws ParseException {
        StringBuilder sql = new StringBuilder();
        sql.append("select ");
        sql.append("e.event_id as eventId ,");
        sql.append("e.project_id as projectId ,");
        sql.append("e.app_name as projectName ,");
        sql.append("e.location_code as locationCode ,");
        sql.append("e.biz_id as bizId ,");
        sql.append("e.biz_type as bizType ,");
        sql.append("e.event_type as eventType ,");
        sql.append("e.creator as creator ,");
        sql.append("r.status as status ,");
        sql.append("r.update_time as updateTime ,");
        sql.append("r.stack_context as stackContext ");
        if (null != req.getStartTime() && DateUtils.addDays(DateUtils.parseDate(req.getStartTime(), YYYY_MM_DD_00_00_00), beforeDays)
                .before(DateHelper.getDateStart(new Date()))) {
            sql.append("from biz_event_bak e  left join biz_event_result_bak r on e.topic=r.topic and e.event_id = r.event_id ");
        } else {
            sql.append("from biz_event e  left join biz_event_result r on e.topic=r.topic and e.event_id = r.event_id ");
        }
        sql.append("where ");
        sql.append("e.project_id = ").append(req.getProjectId());
        if (!StringUtils.isEmpty(req.getEventType())) {
            sql.append(" and e.event_type = ").append("'").append(req.getEventType()).append("' ");
        }
        if (null != req.getStatus()) {
            sql.append(" and r.status = ").append(req.getStatus()).append(" ");
        }
        if (!StringUtils.isEmpty(req.getLocationCode())) {
            sql.append(" and e.location_code = ").append("'").append(req.getLocationCode()).append("' ");
        }
        if (!StringUtils.isEmpty(req.getBizId())) {
            sql.append(" and e.biz_id = ").append("'").append(req.getBizId()).append("' ");
        }
        if (!StringUtils.isEmpty(req.getStartTime()) && !StringUtils.isEmpty(req.getEndTime())) {
            sql.append(" and e.create_time >= ").append("'").append(req.getStartTime()).append("' ");
            sql.append(" and e.create_time <= ").append("'").append(req.getEndTime()).append("' ");
        }
        PageRequest page = PageRequest.of(req.getPage() - 1, req.getSize());
        sql.append(" limit ").append(page.getOffset()).append(",").append(req.getSize());
        log.info("=sql=>sql:{}", sql.toString());
        return sql.toString();
    }


    public String countSql(EventReq req) throws ParseException {
        StringBuilder sql = new StringBuilder();
        sql.append("select count(1) as count ");
        if (null != req.getStartTime() && DateUtils.addDays(DateUtils.parseDate(req.getStartTime(), YYYY_MM_DD_00_00_00), beforeDays)
                .before(DateHelper.getDateStart(new Date()))) {
            sql.append("from biz_event_bak e  left join biz_event_result_bak r on e.topic=r.topic and e.event_id = r.event_id ");
        } else {
            sql.append("from biz_event e  left join biz_event_result r on e.topic=r.topic and e.event_id = r.event_id ");
        }
        sql.append("where ");
        sql.append("e.project_id = ").append(req.getProjectId());
        if (!StringUtils.isEmpty(req.getEventType())) {
            sql.append(" and e.event_type = ").append("'").append(req.getEventType()).append("' ");
        }
        if (null != req.getStatus()) {
            sql.append(" and r.status = ").append(req.getStatus()).append(" ");
        }
        if (!StringUtils.isEmpty(req.getLocationCode())) {
            sql.append(" and e.location_code = ").append("'").append(req.getLocationCode()).append("' ");
        }
        if (!StringUtils.isEmpty(req.getBizId())) {
            sql.append(" and e.biz_id = ").append("'").append(req.getBizId()).append("' ");
        }
        if (!StringUtils.isEmpty(req.getStartTime()) && !StringUtils.isEmpty(req.getEndTime())) {
            sql.append(" and e.create_time >= ").append("'").append(req.getStartTime()).append("' ");
            sql.append(" and e.create_time <= ").append("'").append(req.getEndTime()).append("' ");
        }
        log.info("=sql=>sql:{}", sql.toString());
        return sql.toString();
    }

}
